package report

import (
	"sync"
	"sync/atomic"
	"time"

	"gddgame.cc/galaxy/asteroid"
	"gddgame.cc/galaxy/binocle/logic"
	"gddgame.cc/galaxy/utils"
	"gddgame.cc/galaxy/utils/def"
	"gddgame.cc/galaxy/utils/uuid"
)

type AgentReporter interface {
	AddMetrics() *MetricBuilder
	AddMetric(metric Metric, exec MetricFunc)
}

type MetricReporter struct {
	agent *agentReport

	builder *MetricBuilder

	info   sync.Map
	result map[string]float32
	l      int
}

func (reporter *MetricReporter) init() {
	reporter.info = sync.Map{}
	reporter.l = 0

	t := reporter.builder.getType()
	p := reporter.builder.getPeriod()
	// 添加mock，用于定时调度collect函数
	reporter.agent.AddMetric(Metric{
		Key:    uuid.RandomString(),
		Type:   t,
		Period: p,
		mock:   true,
	}, func(kk string) float32 {
		reporter.result = make(map[string]float32, reporter.l)
		reporter.info.Range(func(key, value interface{}) bool {
			k := utils.ToString(key)
			reporter.result[k] = float32(*value.(*int32))
			// 重置
			atomic.StoreInt32(value.(*int32), 0)

			if !reporter.agent.HasMetric(k) {
				reporter.Register(k)
			}
			return true
		})
		return 0.0
	})
}

func (reporter *MetricReporter) load(key string) float32 {
	if v, ok := reporter.result[key]; ok {
		return v
	} else {
		return 0.0
	}
}

func (reporter *MetricReporter) Register(keys ...string) {
	t := reporter.builder.getType()
	p := reporter.builder.getPeriod()
	for _, key := range keys {
		reporter.agent.AddMetric(Metric{
			Key:       key,
			Type:      t,
			Period:    p,
			NodeLogic: reporter.builder.getNodeLogic(key),
			TimeLogic: reporter.builder.getTimeLogic(key),
		}, reporter.load)
		reporter.l += 1
	}
}

func (reporter *MetricReporter) Increase(key string) {
	act, _ := reporter.info.LoadOrStore(key, utils.Int32(0))
	atomic.AddInt32(act.(*int32), 1)
}

func (reporter *MetricReporter) Decrease(key string) {
	act, _ := reporter.info.LoadOrStore(key, utils.Int32(0))
	atomic.AddInt32(act.(*int32), -1)
}

type MetricBuilder struct {
	agent         *agentReport
	parentBuilder *MetricBuilder

	metricType   MetricType
	period       MetricPeriod
	node         MetricLogic
	time         MetricLogic
	aggregateMap map[string][]MetricLogic
}

func (builder *MetricBuilder) Inherit() *MetricBuilder {
	return &MetricBuilder{agent: builder.agent, parentBuilder: builder}
}

func (builder *MetricBuilder) Type(metricType MetricType) *MetricBuilder {
	builder.metricType = metricType
	return builder
}

func (builder *MetricBuilder) Period(period MetricPeriod) *MetricBuilder {
	builder.period = period
	return builder
}

func (builder *MetricBuilder) Aggregate(node MetricLogic, time MetricLogic) *MetricBuilder {
	builder.node = node
	builder.time = time
	return builder
}

func (builder *MetricBuilder) AggregateSingle(key string, node MetricLogic, time MetricLogic) *MetricBuilder {
	if builder.aggregateMap == nil {
		builder.aggregateMap = make(map[string][]MetricLogic)
	}
	builder.aggregateMap[key] = []MetricLogic{node, time}
	return builder
}

func (builder *MetricBuilder) Exec(key string, exec MetricFunc) *MetricBuilder {
	builder.agent.AddMetric(Metric{
		Key:       key,
		Type:      builder.getType(),
		Period:    builder.getPeriod(),
		NodeLogic: builder.getNodeLogic(key),
		TimeLogic: builder.getTimeLogic(key),
	}, exec)
	return builder
}

func (builder *MetricBuilder) Collect(collect MetricsFunc) *MetricBuilder {
	t := builder.getType()
	p := builder.getPeriod()
	var info map[string]float32
	inlineLoad := func(key string) float32 {
		if v, ok := info[key]; ok {
			return v
		} else {
			return 0.0
		}
	}
	tmp := map[string]struct{}{}
	// 添加mock，用于定时调度collect函数
	builder.agent.AddMetric(Metric{
		Key:    uuid.RandomString(),
		Type:   t,
		Period: p,
		mock:   true,
	}, func(k string) float32 {
		info = collect()
		for key, _ := range info {
			if _, ok := tmp[key]; ok {
				continue
			}
			tmp[key] = struct{}{}
			builder.agent.AddMetric(Metric{
				Key:       key,
				Type:      t,
				Period:    p,
				NodeLogic: builder.getNodeLogic(key),
				TimeLogic: builder.getTimeLogic(key),
			}, inlineLoad)
		}
		return 0.0
	})
	return builder
}

func (builder *MetricBuilder) Reporter() *MetricReporter {
	reporter := &MetricReporter{
		agent:   builder.agent,
		builder: builder,
	}
	reporter.init()
	return reporter
}

func (builder *MetricBuilder) getType() MetricType {
	if builder.metricType != allMetric {
		return builder.metricType
	} else if builder.parentBuilder != nil {
		return builder.parentBuilder.getType()
	} else {
		return MemoryMetric
	}
}

func (builder *MetricBuilder) getPeriod() MetricPeriod {
	if builder.period != 0 {
		return builder.period
	} else if builder.parentBuilder != nil {
		return builder.parentBuilder.getPeriod()
	} else {
		return Minute1
	}
}

func (builder *MetricBuilder) getNodeLogic(key string) MetricLogic {
	if builder.aggregateMap != nil {
		if value, ok := builder.aggregateMap[key]; ok {
			return value[0]
		}
	}
	if builder.node != 0 {
		return builder.node
	} else if builder.parentBuilder != nil {
		return builder.parentBuilder.getNodeLogic(key)
	} else {
		return Sum
	}
}

func (builder *MetricBuilder) getTimeLogic(key string) MetricLogic {
	if builder.aggregateMap != nil {
		if value, ok := builder.aggregateMap[key]; ok {
			return value[1]
		}
	}
	if builder.time != 0 {
		return builder.time
	} else if builder.parentBuilder != nil {
		return builder.parentBuilder.getTimeLogic(key)
	} else {
		return Max
	}
}

type agentReport struct {
	observer logic.Observer

	metricMap sync.Map
	ticker    *time.Ticker
}

func newAgentReport(observer logic.Observer, subscriber asteroid.Sub) *agentReport {
	r := &agentReport{
		observer: observer,

		metricMap: sync.Map{},
	}
	// 按最小频次调度
	//go func() {
	//	now := time.Now()
	//	// 冗余1秒时间差，接收端向下匹配时间
	//	<-time.After(time.Second * time.Duration(61-now.Second()))
	//	ticker := time.NewTicker(time.Second * 30)
	//	r.ticker = ticker
	//	r.onReport()
	//	for _ = range ticker.C {
	//		r.onReport()
	//	}
	//}()
	subscriber.On("report/request", func(message asteroid.Message) {
		metric := message.GetTarget()
		if m, ok := r.metricMap.Load(metric); ok {
			message.Response(m)
		} else {
			message.Response(false)
		}
	})
	return r
}

func (r *agentReport) onReport() {
	now := time.Now()
	result := []interface{}{}
	r.metricMap.Range(func(key, value interface{}) bool {
		metric := value.(*Metric)
		if (metric.lastTime == nil && now.Sub(now.Truncate(metric.duration)) < time.Second*30) || (metric.lastTime != nil && now.Sub(*metric.lastTime) > metric.duration) {
			value := metric.exec(metric.Key)
			if metric.mock {
				return true
			}
			result = append(result, key, value)
			metric.lastTime = &now
		}
		return true
	})
	if len(result) == 0 {
		return
	}
	r.observer.PublishPipe().List(result...).Send("report")
}

func (r *agentReport) AddMetrics() *MetricBuilder {
	return &MetricBuilder{agent: r}
}

func (r *agentReport) AddMetric(metric Metric, exec MetricFunc) {
	r.metricMap.Store(metric.Key, &metric)
	metric.Init(exec)
}

func (r *agentReport) HasMetric(key string) bool {
	_, ok := r.metricMap.Load(key)
	return ok
}

const (
	initLength       = 30
	maxHistoryLength = 120
	secondFormat     = "2006-01-02 15:04:05"
)

type ReportStore interface {
	GetReports(observer string, metric string, startTime *time.Time, endTime *time.Time) []MetricReport
	SaveReports(observer string, metric string, value float32, t *time.Time)
}

type CoreReporter interface {
	// 基础统计
	NodeSourceReports(node string, metrics []string) [][]interface{}
	// 设定cycle的统计：内存记录计算
	NodeReports(node string, metrics []string, cycle MetricCycle, duration time.Duration) [][]interface{}
	// 服务为单位，内存记录计算
	Reports(observer string, metrics []string, cycle MetricCycle, duration time.Duration) [][]interface{}
	// 服务为单位: 支持1小时以上查询逻辑，读取数据库
	ReportsWithTime(observer string, metrics []string, cycle MetricCycle, startTime *time.Time, endTime *time.Time) [][]interface{}
	BindReportStore(store ReportStore)
	// 获取服务所有统计项
	Metrics(observer string, t MetricType) []string
}

type MetricReport struct {
	Time  *time.Time
	Value float32
}
type nodeReport struct {
	Metrics  sync.Map // map[string][]MetricReport
	Observer *observerReport
}
type observerReport struct {
	Metrics sync.Map // map[string]*Metric
}
type coreReport struct {
	watcher   *logic.WatcherInstance
	observers sync.Map
	nodes     sync.Map
	store     ReportStore
	ticker    *time.Ticker
}

func newCoreReport(watcher *logic.WatcherInstance, system logic.System) *coreReport {
	r := &coreReport{
		watcher:   watcher,
		nodes:     sync.Map{},
		observers: sync.Map{},
	}
	// 小时数据统计合并
	go func() {
		now := time.Now()
		// 冗余5秒时间差，等待数据发送
		<-time.After(time.Minute*time.Duration(60-now.Minute()) + time.Second*time.Duration(65-now.Second()))
		ticker := time.NewTicker(time.Hour)
		r.ticker = ticker
		r.onReportHourly()
		for _ = range ticker.C {
			r.onReportHourly()
		}
	}()
	watcher.Watching("report", func(message asteroid.Message) {
		observer := message.GetOrigin()
		node := message.GetFrom()
		r.appendReport(observer, node, message.List())
	})
	system.OnFinish(func() {
		// 将重新分配的服务数据删除
		r.observers.Range(func(key, value interface{}) bool {
			observer := key.(string)
			node := watcher.AllocObserver(observer)
			if node != system.Node() {
				r.observers.Delete(key)
				nodes := watcher.NodeList(observer)
				for _, node := range nodes {
					r.nodes.Delete(node)
				}
			}
			return true
		})
	})
	// 自动心跳report
	watcher.OnAlive(func(observer string, node string, online bool) {
		if watcher.AllocObserver(observer) != system.Node() {
			return
		}
		value := 0
		if online {
			value = 1
		}
		r.appendReport(observer, node, []interface{}{"alive", value})
	})
	return r
}

func (r *coreReport) getObserver(observer string) *observerReport {
	var or *observerReport
	o, ok := r.observers.Load(observer)
	if !ok {
		or = &observerReport{
			Metrics: sync.Map{}, //make(map[string]*Metric),
		}
		// 预设alive
		or.Metrics.Store("alive", &Metric{
			Key:       "alive",
			Period:    Second30,
			Type:      MemoryMetric,
			NodeLogic: Sum,
			TimeLogic: Min,
		})
		r.observers.Store(observer, or)
	} else {
		or = o.(*observerReport)
	}
	return or
}

func (r *coreReport) BindReportStore(store ReportStore) {
	r.store = store
}

func (r *coreReport) appendReport(observer string, node string, data []interface{}) {
	var nr *nodeReport
	var or *observerReport
	s, ok := r.nodes.Load(node)
	if !ok {
		nr = &nodeReport{
			Metrics: sync.Map{}, // make(map[string][]MetricReport),
		}
		r.nodes.Store(node, nr)
		or := r.getObserver(observer)
		nr.Observer = or
	} else {
		nr = s.(*nodeReport)
	}
	or = nr.Observer
	now := time.Now()
	now = now.Truncate(time.Second * 30)

	result := make(map[string]float32)
	for i := 0; i < len(data); i += 2 {
		m := utils.ToString(data[i])
		value := utils.ToFloat32(data[i+1])
		result[m] = value
		var reports []MetricReport
		rr, ok := nr.Metrics.Load(m)
		if !ok {
			reports = make([]MetricReport, 0, initLength)
		} else {
			reports = rr.([]MetricReport)
		}
		if _, ok := or.Metrics.Load(m); !ok {
			r.requestMetric(observer, node, m)
		}
		if len(reports) == maxHistoryLength {
			reports = reports[1:]
		}

		nr.Metrics.Store(m, append(reports, MetricReport{
			Time:  &now,
			Value: value,
		}))

	}
}

func (r *coreReport) onReportHourly() {
	now := time.Now()
	now = now.Truncate(time.Hour)
	// 小时
	start := now.Add(-time.Hour)
	r.observers.Range(func(key, value interface{}) bool {
		or := value.(*observerReport)
		observer := key.(string)
		nodes := r.watcher.NodeList(observer)
		or.Metrics.Range(func(key, value interface{}) bool {
			metric := value.(*Metric)
			if metric.Type&PersistenceMetric == 0 {
				return true
			}
			nodeHandler := r.getHandler(metric.NodeLogic)
			timeHandler := r.getHandler(metric.TimeLogic)
			values := make([]float32, 0, len(nodes))
			for _, node := range nodes {
				s, ok := r.nodes.Load(node)
				if !ok {
					continue
				}
				nodeValues := make([]float32, 0, maxHistoryLength)
				report := s.(*nodeReport)
				mm, ok := report.Metrics.Load(metric.Key)
				if !ok {
					continue
				}
				mr := mm.([]MetricReport)
				for k := range mr {
					if mr[k].Time.After(start) && mr[k].Time.Before(now) {
						nodeValues = append(nodeValues, mr[k].Value)
					}
				}
				values = append(values, timeHandler(values))
			}
			if r.store != nil {
				r.store.SaveReports(observer, metric.Key, nodeHandler(values), &now)
			}
			return true
		})
		return true
	})
}

func (r *coreReport) NodeSourceReports(node string, metrics []string) [][]interface{} {
	s, ok := r.nodes.Load(node)
	if !ok {
		return nil
	}
	report := s.(*nodeReport)
	result := make([][]interface{}, len(metrics))
	for index, metricKey := range metrics {
		mm, ok := report.Metrics.Load(metricKey)
		if !ok {
			continue
		}
		mr := mm.([]MetricReport)
		result[index] = r.formatOut(mr)
	}
	return result
}

func (r *coreReport) NodeReports(node string, metrics []string, cycle MetricCycle, duration time.Duration) [][]interface{} {
	s, ok := r.nodes.Load(node)
	if !ok {
		return nil
	}
	d := r.getDuration(cycle)
	startTime := time.Now().Add(-1 * duration)
	report := s.(*nodeReport)
	result := make([][]interface{}, len(metrics))
	for index, metricKey := range metrics {
		mm, ok := report.Metrics.Load(metricKey)
		if !ok {
			continue
		}
		m, ok := report.Observer.Metrics.Load(metricKey)
		if !ok {
			continue
		}
		metric := m.(*Metric)
		mr := mm.([]MetricReport)

		mr = r.mergeDuration(mr, metric.TimeLogic, &startTime, d)
		result[index] = r.formatOut(mr)
	}
	return result
}

func (r *coreReport) Reports(observer string, metrics []string, cycle MetricCycle, duration time.Duration) [][]interface{} {
	d := r.getDuration(cycle)
	startTime := time.Now().Add(-1 * duration)
	result := make([][]interface{}, len(metrics))
	or := r.getObserver(observer)
	nodes := r.watcher.NodeList(observer)
	//fmt.Println(observer, startTime, nodes)
	for index, metricKey := range metrics {
		value, ok := or.Metrics.Load(metricKey)
		//fmt.Println(metricKey, ok, value)
		if !ok {
			result[index] = []interface{}{}
			continue
		}
		metric := value.(*Metric)
		values := make([][]MetricReport, 0, len(nodes))
		for _, node := range nodes {
			s, ok := r.nodes.Load(node)
			if !ok {
				continue
			}
			report := s.(*nodeReport)
			mm, ok := report.Metrics.Load(metric.Key)
			//fmt.Println(metricKey, node, ok, mm)
			if !ok {
				continue
			}
			mr := mm.([]MetricReport)
			mr = r.mergeDuration(mr, metric.TimeLogic, &startTime, d)
			values = append(values, mr)
		}
		result[index] = r.formatOut(r.mergeNode(values, metric.NodeLogic, &startTime, d))
	}
	return result
}

func (r *coreReport) ReportsWithTime(observer string, metrics []string, cycle MetricCycle, startTime *time.Time, endTime *time.Time) [][]interface{} {
	d := r.getDuration(cycle)
	result := make([][]interface{}, len(metrics))
	or := r.getObserver(observer)
	for index, metricKey := range metrics {
		mr := r.store.GetReports(observer, metricKey, startTime, endTime)
		value, ok := or.Metrics.Load(metricKey)
		if !ok {
			result[index] = []interface{}{}
			continue
		}
		metric := value.(*Metric)
		mr = r.mergeDuration(mr, metric.TimeLogic, startTime, d)
		result[index] = r.formatOut(mr)
	}
	return result
}

func (r *coreReport) Metrics(observer string, t MetricType) []string {
	s, ok := r.observers.Load(observer)
	if !ok {
		return nil
	}
	or := s.(*observerReport)
	result := make([]string, 0, 5)
	or.Metrics.Range(func(key, value interface{}) bool {
		metric := value.(*Metric)
		if t == allMetric || metric.Type&t > 0 {
			result = append(result, metric.Key)
		}
		return true
	})
	return result
}

func (r *coreReport) requestMetric(observer string, node string, metric string) {
	publisher := r.watcher.Observing(observer)
	if publisher != nil {
		publisher.Direct(node).TargetId(metric).ResponseSerialize(func() def.OptimizeSerializer {
			return &Metric{}
		}).CallbackTo(func(success bool, data interface{}) {
			if success {
				or := r.getObserver(observer)
				switch v := data.(type) {
				case *Metric:
					or.Metrics.Store(metric, v)
				default:
				}
			}
		}).Send("report/request")
	}
}

func (r *coreReport) getHandler(metricLogic MetricLogic) MetricLogicHandler {
	switch metricLogic {
	case Max:
		return MaxHandle
	case Min:
		return MinHandle
	case Avg:
		return AvgHandle
	case Sum:
		return SumHandle
	}
	return nil
}

func (r *coreReport) formatOut(mr []MetricReport) []interface{} {
	rr := make([]interface{}, len(mr)*2)
	for i := range mr {
		formatTime := mr[i].Time.Format(secondFormat)
		rr[i*2] = formatTime
		rr[i*2+1] = mr[i].Value
	}
	return rr
}

// 按时间段合并数据
func (r *coreReport) mergeDuration(mr []MetricReport, metricLogic MetricLogic, startTime *time.Time, duration time.Duration) []MetricReport {
	handler := r.getHandler(metricLogic)
	tmp := make([]MetricReport, 0, 10)
	values := make([]float32, 0, 24)
	start := startTime.Truncate(duration).Add(-1 * time.Second)
	end := start.Add(duration)
	//fmt.Println(start, end)
	for k := range mr {
		if mr[k].Time.Before(start) {
			continue
		} else if mr[k].Time.Before(end) {
			values = append(values, mr[k].Value)
		} else {
			t := start.Add(time.Second)
			tmp = append(tmp, MetricReport{
				Value: handler(values),
				Time:  &t,
			})
			values = values[:0]
			start = end
			end = end.Add(duration)
		}
	}
	if len(values) > 0 {
		t := start.Add(time.Second)
		tmp = append(tmp, MetricReport{
			Value: handler(values),
			Time:  &t,
		})
	}
	//fmt.Println(tmp)
	return tmp
}

// 按节点合并相同时间段数据
func (r *coreReport) mergeNode(mrs [][]MetricReport, nodeLogic MetricLogic, startTime *time.Time, duration time.Duration) []MetricReport {
	handler := r.getHandler(nodeLogic)
	tmp := make([]MetricReport, 0, 10)
	values := make([]float32, 0, len(mrs))

	// 保存索引
	indexs := make([]int, len(mrs))
	for k := range mrs {
		indexs[k] = -1
	}

	start := startTime.Truncate(duration).Add(-1 * time.Second)
	end := start.Add(duration)
	//fmt.Println(start, end)
D:
	for index := range mrs {
		mr := mrs[index]
		k := indexs[index] + 1
		for ; k < len(mr); k++ {
			// 查找该时间段report
			if mr[k].Time.Before(start) {
				continue
			} else if mr[k].Time.Before(end) {
				values = append(values, mr[k].Value)
				break
			} else {
				// 有数据丢失,保持索引
				k -= 1
				break
			}
		}
		indexs[index] = k
	}
	if len(values) > 0 {
		t := start.Add(time.Second)
		tmp = append(tmp, MetricReport{
			Value: handler(values),
			Time:  &t,
		})
		values = values[:0]
		start = end
		end = end.Add(duration)
		goto D
	}

	//fmt.Println(tmp)
	return tmp
}

func (r *coreReport) getDuration(cycle MetricCycle) time.Duration {
	switch cycle {
	case Minute1Cycle:
		return time.Minute
	case Minute5Cycle:
		return time.Minute * 5
	case Minute15Cycle:
		return time.Minute * 15
	case Minute30Cycle:
		return time.Minute * 30
	case Hour1Cycle:
		return time.Hour
	case Hour12Cycle:
		return time.Hour * 12
	case Hour24Cycle:
		fallthrough
	default:
		return time.Hour * 24
	}
}
